#include <deque>
#include <algorithm>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include "pwext.hpp"

struct Message 
{
	Message(int mtype) : msgtype_(mtype)
	{ }
	Message()
	{ }
	virtual ~Message() { } // allow for polymorphism

	enum {
		EXIT,
		CANCEL,
		NORMAL, // first normal type
		SOME_MSG_1,
		DUMMY
	};
	int msgtype_;
	// ... whatever else is needed by the application ...
};

// copied from ex04dc
class Queue
{
protected:
	enum post_l_result {
		P_FAIL,
		P_SUCCESS,
		P_POSTIT
	};

public:
	enum Priority {
		PRIORITY_HIGH = 0,
		PRIORITY_NORMAL // all the other values are treated as NORMAL
	};

	enum {
		DEFAULT_LIMIT = 1000
	};

	// use the AllExitQueue approach, with only the "forwards" connections
	// counted for the posting of EXIT
	Queue(int limit = DEFAULT_LIMIT) :
		limit_(limit), 
		condrd_(condwr_), priority_not_empty_(false),
		exitposted_(false), nwriters_(0)
	{ 
		int i = 0;
		dqwr_[PRIORITY_HIGH] = &deques_[i++];
		dqrd_[PRIORITY_HIGH] = &deques_[i++];
		dqwr_[PRIORITY_NORMAL] = &deques_[i++];
		dqrd_[PRIORITY_NORMAL] = &deques_[i++];
	}

	virtual ~Queue()
	{ 
		Message *m;
		for (int i = 0; i < 4; i++) {
			while (!deques_[i].empty()) {
				m = deques_[i].front();
				deques_[i].pop_front();
				delete m;
			}
		}
	}

	// virtual in case if later a subclass wants to do something different;
	// returns true on success, or false if the handler has already been told
	// to exit
	virtual bool add_writer()
	{ 
		pw::lockmutex lm(mutexx_);
		if (exitposted_)
			return false;
		++nwriters_;
		return true;
	}

	// this combines the STL front() and pop_front() since
	// the popping must be atomic;
	// the receiver takes over the ownership of the message
	Message *pop_front()
	{
		pw::lockmutex lm(mutexrd_);

		// if the reading of NORMAL dqrd was in progress but
		// a PRIORITY message arrived, must switch to reading
		// the PRIORITY messages.
		if (priority_not_empty_ && dqrd_[PRIORITY_HIGH]->empty()) {
			swap(dqwr_[PRIORITY_HIGH], dqrd_[PRIORITY_HIGH]);
			priority_not_empty_ = false;
		}

		while(dqrd_[PRIORITY_HIGH]->empty()
		&& dqrd_[PRIORITY_NORMAL]->empty()) {
			pw::lockmutex lm(condwr_);

			if (dqwr_[PRIORITY_HIGH]->empty()
			&& dqwr_[PRIORITY_NORMAL]->empty()) {
				condrd_.wait(); // it releases and gets back the mutex
			} else {
				if (!dqwr_[PRIORITY_HIGH]->empty()) {
					swap(dqwr_[PRIORITY_HIGH], dqrd_[PRIORITY_HIGH]);
					// high-priority posting doesn't block, 
					// no need to wake up the writers
				}
				if (!dqwr_[PRIORITY_NORMAL]->empty()) {
					swap(dqwr_[PRIORITY_NORMAL], dqrd_[PRIORITY_NORMAL]);
					// if the write buffer was full, wake up the writers
					condwr_.broadcast();
				}
			}
		}

		Message *m;
		if(!dqrd_[PRIORITY_HIGH]->empty()) {
			m = dqrd_[PRIORITY_HIGH]->front();
			dqrd_[PRIORITY_HIGH]->pop_front();
		} else {
			m = dqrd_[PRIORITY_NORMAL]->front();
			dqrd_[PRIORITY_NORMAL]->pop_front();
		}
		return m;
	}

	// takes over the ownership of the message;
	// returns true if the message was posted, false if discarded
	virtual bool post(Priority pri, Message *m)
	{
		if (pri != PRIORITY_HIGH)
			pri = PRIORITY_NORMAL;

		int res;
		{
			pw::lockmutex lm(mutexx_);
			res = post_l(pri, m);
		}
		if (res == P_POSTIT)
			return push_back(pri, m);
		else
			return (res != P_FAIL);
	}

protected:

	virtual post_l_result post_l(Priority pri, Message *m)
	{
		if (exitposted_) {
			delete m;
			return P_FAIL;
		}
		if (m->msgtype_ == Message::EXIT) {
			if (pri == PRIORITY_HIGH) {
				delete m;
				return P_FAIL; // can't send EXIT in a priority message
			}

			if (--nwriters_ <= 0) {
				// last writer, post EXIT to the handler
				exitposted_ = true;
			} else {
				// message is processed successfully but it has
				// nothing to pass to the handler yet
				delete m;
				return P_SUCCESS;
			}
		}
		return P_POSTIT;
	}

	// returns false if the exit was already posted, and this
	// is not the EXIT message itself (then the message also
	// gets discarded);
	// expects that the priority value is in the correct range
	bool push_back(Priority pri, Message *m)
	{
		bool need_prodding, result;
		{
			pw::lockmutex lm(condwr_);
			result = push_back_l(pri, m, need_prodding);
		}
		if (need_prodding) {
			pw::lockmutex lm(mutexrd_);
			priority_not_empty_ = true;
		}
		return result;
	}
	// Caller must have the mutex in condwr_ locked. 
	// If need_prodding is returned true, the caller must set
	// the flag priority_not_empty_ with locked mutexrd_.
	bool push_back_l(Priority pri, Message *m, bool &need_prodding)
	{
		need_prodding = false;
		if (exitposted_ && m->msgtype_ != Message::EXIT) {
			delete m;
			return false;
		}

		if (pri == PRIORITY_HIGH) {
			// remember if the reader may need prodding to
			// switch from the buffer of normal messages to
			// priority ones
			need_prodding = dqwr_[PRIORITY_HIGH]->empty();
		} else {
			// flow control only for the normal messages
			while (dqwr_[PRIORITY_NORMAL]->size() >= limit_)
				condwr_.wait(); // it releases and gets back the mutex
		}

		dqwr_[pri]->push_back(m);
		condrd_.broadcast();
		return true;
	}

	int limit_;
	std::deque<Message *> deques_[4];
	pw::pmutex mutexrd_;
	std::deque<Message *> *dqwr_[2], *dqrd_[2];
	pw::pmcond condwr_;
	pw::pchaincond condrd_;
	bool priority_not_empty_; // flag: the priority dqwr became non-empty
	pw::pmutex mutexx_; // for the exit accounting
	bool exitposted_; // whether the EXIT message was posted to the queue
	int nwriters_; // number of writers that haven't already exited
};

class Modifier;

// [[ex04di]] 
// DEBUG descr Sending the notifications and waiting for them in the same thread causes a deadlock.
class DummyMessage : public Message
{
public:
	DummyMessage(Modifier *mod) :
		Message(), mod_(mod), chained_(0)
	{ 
		msgtype_ = DUMMY;
	}

	// DEBUG getfrom DummyMessage::~DummyMessage
	// DEBUG ~DummyMessage definition moved to after Modifier
	~DummyMessage();
	// DEBUG endget

	// DEBUG getfrom DummyMessage::chain
	// DEBUG chain definition moved to after Modifier
	Message *chain();
	// DEBUG endget

	void completed(Message *m)
	{
		chained_ = m;
		done_.signallock();
	}

protected:
	Modifier *mod_;
	// for passing between completed() and chain()
	Message *chained_; 
	pw::pmcond done_;
};

class Modifier
{
	friend class DummyMessage;
public:
	Modifier(Queue *q) : dummy_(0), q_(q) 
	{
		// ...
	}

	void execute() 
	{
		// ...
		// now it's time to do the modification
		dummy_ = new DummyMessage(this);
		{
			pw::lockmutex lm(ready_);
			q_->post(Queue::PRIORITY_NORMAL, dummy_);
			ready_.wait();
		}
		// after wait succeeded, dummy may be accessed
		// again

		if (dummy_) { 
			// ... calculate the modification, build the
			// message in m....
	// DEBUG {
			fprintf(stderr, "Modifier starts\n");
			Message *m = new Message(Message::NORMAL);
	// DEBUG }
			dummy_->completed(m);
		} else {
			// ... do the clean-up if the modifier was
			// skipped
		}
		// after this point dummy cannot be accessed by
		// the modifier ...
	}

	void dummy_signal()
	{
		ready_.signallock();
	}

	void dummy_discard()
	{
		dummy_ = 0;
		ready_.signallock();
	}

protected:
	pw::pmcond ready_;
	DummyMessage *dummy_;
	Queue *q_;
};
// DEBUG move DummyMessage::~DummyMessage
	/*DEBUG{*/DummyMessage::/*}DEBUG*/~DummyMessage() 
	{ 
		if (mod_) {
			mod_->dummy_discard();
		}
	}
// DEBUG endmove DummyMessage::~DummyMessage
// DEBUG move DummyMessage::chain
	Message */*DEBUG{*/DummyMessage::/*}DEBUG*/chain()
	{
		if (mod_) {
			pw::lockmutex lm(done_);
			mod_->dummy_signal();
			done_.wait();
			mod_ = 0;
		}
		return chained_;
	}
// DEBUG endmove DummyMessage::chain
// END

// DEBUG {
class Handler : public pw::pwthread
{
public:
	void setMyQueue(Queue *q)
	{
		q_ = q;
	}
	void *execute()
	{
		while(true) {
			Message *m = q_->pop_front();
			if (m->msgtype_ == Message::DUMMY) {
				fprintf(stderr, "Handling the modifier message\n");
				Message *newm = ((DummyMessage *)m)->chain();
				delete m;
				m = newm;
				if (newm == 0)
					continue;
			}
			switch(m->msgtype_) {
			case Message::EXIT:
				fprintf(stderr, "Handling the exit message\n");
				delete m;
				return 0;
			default:
				fprintf(stderr, "Handling the common message\n");
				delete m;
				break;
				// ...
			}
		}
		return 0;
	}
protected:
	Queue *q_;
};



int main()
{
	Handler h;
	Queue q;
	q.add_writer();

	h.setMyQueue(&q);

	h.start();

	Modifier mod(&q);
	mod.execute();

	q.post(Queue::PRIORITY_NORMAL, new Message(Message::EXIT));
	h.join();

	mod.execute();


	return 0;
}
/* Sample output (timing of URGENT processing may vary):
Process URGENT_1 3
deleted data 3
Start process normal 1
End process normal 1
deleted data 1
Start process normal 2
End process normal 2
deleted data 2
Process exit
deleted data 99
*/
// DEBUG }
